package com.celeritymq.connection;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.SocketAddress;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.ConnectionMetaData;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;

import com.celeritymq.listeners.CelerityExceptionListener;
import com.celeritymq.utils.CelerityObject;

public class CelerityConnectionImpl implements CelerityConnection, CelerityObject {

	private volatile AtomicBoolean start = new AtomicBoolean(false);

	private String clientId;
	private CelerityExceptionListener celerityExceptionListener;
	private CelerityConnectionMetaData connectionMetaData;
	private Socket socket;
	protected SocketAddress targetAddress;
	private Reader reader;
	private Writer writer;
	private BlockingQueue<Message> inputQueue;
	private BlockingQueue<Message> outputQueue;
	private CelerityMode mode;
	private int bufferSize;

	public CelerityConnectionImpl(SocketAddress address, int bufferSize) throws IOException {
		this.targetAddress = address;
		mode = new CelerityMode(CelerityMode.NONE);
		socket = new Socket();
		this.bufferSize = bufferSize;
	}

	public void register(int mode) {
		this.mode.add(mode);
	}

	public void unRegister(int mode) {
		this.mode.remove(mode);
	}

	@Override
	public String getClientID() throws JMSException {
		return clientId;
	}

	@Override
	public ExceptionListener getExceptionListener() throws JMSException {
		return celerityExceptionListener;
	}

	@Override
	public ConnectionMetaData getMetaData() throws JMSException {
		return connectionMetaData;
	}

	@Override
	public void setClientID(String clientId) throws JMSException {
		this.clientId = clientId;
	}

	@Override
	public void setExceptionListener(ExceptionListener exceptionListener) throws JMSException {
		throw new UnsupportedOperationException("not support this method");
	}

	public void setCelerityExceptionListener(CelerityExceptionListener celerityExceptionListener) {
		this.celerityExceptionListener = celerityExceptionListener;
	}

	@Override
	public void start() throws JMSException {
		connect();
		if (mode.canRead()) {
			inputQueue = new LinkedBlockingQueue<Message>(bufferSize);
			reader = new Reader();
			reader.work();
		}
		if (mode.canWrite()) {
			outputQueue = new LinkedBlockingQueue<Message>(bufferSize);
			writer = new Writer();
			writer.work();
		}
		start.set(true);
	}

	@Override
	public void stop() throws JMSException {
		start.set(false);
		// if(reader!=null){
		// try {
		// reader.close();
		// } catch (Exception e) {
		// e.printStackTrace();
		// }
		// }
		// if(writer!=null){
		// try {
		// writer.close();
		// } catch (Exception e) {
		// e.printStackTrace();
		// }
		// }
	}

	@Override
	public boolean connect() {
		try {
			socket.connect(targetAddress, 10000);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

	@Override
	public void close() {
		try {
			socket.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean isConnected() {
		return socket.isConnected();
	}

	@Override
	public boolean isClose() {
		return socket.isClosed();
	}

	private class Reader extends Thread {
		private ObjectInputStream ois = null;
		private Thread selfThread;
		private volatile boolean run = false;

		private Reader() {
			try {
				InputStream is = socket.getInputStream();
				ois = new ObjectInputStream(is);
			} catch (Exception e) {
				e.printStackTrace();
				start.set(false);
				return;
			}
		}

		private void work() {
			run = true;
			start();
		}

		public void run() {
			if (ois == null) {
				return;
			}
			selfThread = Thread.currentThread();
			while (start.get()) {
				try {
					Message message = (Message) ois.readObject();
					inputQueue.put(message);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		private void temp() {
			run = false;
		}

		private void close() throws Exception {
			selfThread.interrupt();
			ois.close();
		}
	}

	private class Writer extends Thread {
		private ObjectOutputStream oos;
		private Thread selfThread;

		private Writer() {
			try {
				OutputStream os = socket.getOutputStream();
				oos = new ObjectOutputStream(os);
			} catch (Exception e) {
				try {
					socket.close();
				} catch (IOException e1) {
					e1.printStackTrace();
				}
				start.set(false);
				e.printStackTrace();
			}

		}

		private void work() {
			start();
		}

		public void run() {
			if (oos == null) {
				return;
			}
			selfThread = Thread.currentThread();
			while (start.get()) {
				try {
					Message message = outputQueue.take();
					oos.writeObject(message);
					oos.flush();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}

		private void close() throws Exception {
			selfThread.interrupt();
			oos.close();
		}
	}

	public static void main(String[] args) throws Throwable {
		final BlockingQueue bq = new LinkedBlockingQueue();
		final AtomicBoolean start = new AtomicBoolean(false);
		Thread thread = new Thread() {
			public void run() {
				while (start.get()) {
					System.out.println("start take");
					try {
						bq.take();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("take end");
				}
			}
		};
		start.set(true);
		thread.start();

		TimeUnit.SECONDS.sleep(5);
		start.set(false);
		thread.interrupt();

	}

	@Override
	public CelerityMode getMode() {
		return mode;
	}

	@Override
	public Object createInstance(Object... args) {
		try {
			return new CelerityConnectionImpl((SocketAddress) args[0], (Integer) args[1]);
		} catch (IOException e) {
			e.printStackTrace();
			return null;
		}
	}

}
